Add support for dynamic write in MqttIO
#32470
Conversation
|
Assigning reviewers. If you would like to opt out of this review, comment R: @kennknowles for label java. Available commands:
The PR bot will only process comments in the main thread (not review comments). |
There was a problem hiding this comment.
Thanks for the contribution. The added PTransform and DoFn pretty much duplicated the existing one. I would suggest adjust the current Write transform and WriteFn for this feature. This avoids branches and more independent transform/DoFn to maintain. Should be straightforward to do so.
I see, there was concern of breaking change, Write originally is not a generic but WriteDynamics transform is a generic. We can make Write to return a WriteDynamics<byte[]> with fixed topicFn and identity payloadFn.
@Abacn |
Nope, I just merged conflicted ^_^ and I'm doing working this |
|
@Abacn Thanks for your comment. ^_^ |
| public PDone expand(PCollection<InputT> input) { | ||
| checkArgument(connectionConfiguration() != null, "connectionConfiguration can not be null"); | ||
| final SerializableFunction<InputT, String> topicFn; | ||
| if (dynamic()) { |
There was a problem hiding this comment.
since getTopic() is or not null corresponds to dynamic() being false and true, we can remove this property, and just use topic nullness to determine the scenario?
There was a problem hiding this comment.
This would also allow calls like MqttIO.write().withTopicPayloadFn(...), since they use the same Write class.
We don't know if the Write object was called from write() or dynamicWrite(). This is not enough to determine the scenario based on whether the topic is null.
The topic value can be modified externally, but the dynamic value cannot.
I would like to make sure that withTopicPayloadFn() and withPayloadFn() are only available when dynamic is true. What do you think? and please take a look a the following test code.
@Test
public void testWriteWithTopicFn() {
IllegalArgumentException exception =
assertThrows(
IllegalArgumentException.class, () -> MqttIO.write().withTopicFn(e -> "some topic"));
assertEquals("withTopicFn can not use in non-dynamic write", exception.getMessage());
}
@Test
public void testWriteWithPayloadFn() {
final IllegalArgumentException exception =
assertThrows(
IllegalArgumentException.class, () -> MqttIO.write().withPayloadFn(e -> new byte[] {}));
assertEquals("withPayloadFn can not use in non-dynamic write", exception.getMessage());
}
* add support for dynamic write in MqttIO * Update CHANGES.md * add some assertions in testDynamicWrite * remove whitespace in CHANGES.md * refactor duplicated Write transform * change WriteFn to use Write spec
Please add a meaningful description for your change here
fixes #19376
This PR contains these changes
MqttIOThank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.